--
--  Test walminer extension
--

CREATE EXTENSION walminer;

ALTER SYSTEM SET wal_keep_segments TO 100;
ALTER SYSTEM SET checkpoint_timeout TO 3600;
select pg_reload_conf();

set client_min_messages TO warning;
set extra_float_digits = 0;

-- Simple DML & LSN range test

create table tb1(id int primary key,c1 text);
insert into tb1 values(1,'1');
insert into tb1 values(2,'2');
insert into tb1 values(3,'3');

select pg_current_wal_lsn() lsn_4 \gset
insert into tb1 values(4,'4');

select pg_current_wal_lsn() lsn_5 \gset
insert into tb1 values(5,'5');

insert into tb1 values(6,'6');
update tb1 set c1='2x' where id=2;

select pg_current_wal_lsn() lsn_u3 \gset
update tb1 set id=-3,c1='3x' where id=3;

select pg_current_wal_lsn() lsn_d5 \gset
delete from tb1 where id=5;

select op_type,op_text from wal2sql(NULL,NULL,NULL);

select op_type,op_text from wal2sql(NULL,:'lsn_4',:'lsn_5');

select op_type,op_text from wal2sql(NULL,:'lsn_4',:'lsn_d5');

select op_type,op_text from wal2sql(NULL,:'lsn_u3',:'lsn_d5');

select op_type,op_text from wal2sql(NULL,:'lsn_u3',NULL);

select op_type,op_text from wal2sql(NULL,NULL,:'lsn_d5');

-- All data type test

select pg_current_wal_lsn() lsn_1 \gset

create table tb_types(
id int primary key,
c_bigbit bigint,
c_bigserial bigserial,
c_bit10 bit(10),
c_varbit10 varbit(10),
c_bool bool,
c_box box,
c_bytea bytea,
c_char char,
c_char10 char(10),
c_varchar10 varchar(10),
c_cidr cidr,
c_circle circle,
c_date date,
c_float8 float8,
c_inet inet,
c_int int,
c_interval interval,
c_json json,
c_jsonb jsonb,
c_line line,
c_lseg lseg,
c_macaddr macaddr,
c_macaddr8 macaddr8,
c_money money,
c_numeric10_3 numeric(10,3),
c_path path,
c_pg_lsn pg_lsn,
c_point point,
c_polygon polygon,
c_float4 float4,
c_smallint smallint,
c_smallserial smallserial,
c_serial serial,
c_text text,
c_time  time,
c_timetz timetz,
c_timestamp timestamp,
c_timestamptz timestamptz,
c_tsquery tsquery,
c_tsvector tsvector,
c_txid_snapshot txid_snapshot,
c_uuid uuid,
c_xml xml,
c_inta int[]
);

insert into tb_types values
(
1,
-1,
DEFAULT,
'1000100001',
'1000100001',
true,
'(( 0 , 0 ) , ( 99.9 , 99.9 ))',
'\x616263',
'k',
'abc',
'abc',
'192.168.100.128/25',
'<(0,0),5.1>',
'2019-12-14',
'9.99999',
'192.168.100.128/25',
NULL,
'3 4:05:06'::interval,
'[1, 2, "foo", null]'::json,
'[1, 2, "foo", null]'::jsonb,
'{2,1,100}',
'((0,0),(99.9,99.9))',
'08:00:2b:01:02:03',
'08:00:2b:01:02:03:04:05',
'12.34',
111.333,
'((0,0),(10,10),(15.1,1.1))',
'0/8097378',
'(1.1,99.9)',
'((0,0),(10,10),(15.1,15.1),(0,0))',
99.9,
1,
DEFAULT,
DEFAULT,
'abc123',
'00:04:28.237356',
'00:04:47.28835+08',
'2019-12-14 00:05:21.820347',
'2019-12-14 00:05:21.820347+08',
'fat & cat'::tsquery,
'a fat cat'::tsvector,
'673:673:'::txid_snapshot,
'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'::uuid,
'<foo>bar</foo>'::xml,
'{1,2,4}'::int[]
);

update tb_types
set c_bigbit=2,
c_bigserial=-2,
c_bit10='1000100000',
c_varbit10='1000100000',
c_bool=false,
c_box='(( 1 , 1 ) , ( 99.9 , 99.9 ))'::box,
c_bytea='\x61626364',
c_char='b',
c_char10='abcd',
c_varchar10='abcd',
c_cidr='192.168.101.128/25',
c_circle='<(1,1),5.1>',
c_date='2019-12-15',
c_float8='9.999991',
c_inet='192.168.101.128/25',
c_int=1,
c_interval='3 4:05:07'::interval,
c_json='[1, 2, "foo2", null]'::json,
c_jsonb='[1, 2, "foo2", null]'::jsonb,
c_line='{2,1,100}'::line,
c_lseg='((0,0),(99.9,99.9))'::lseg,
c_macaddr='08:00:2b:01:02:04',
c_macaddr8='08:00:2b:01:02:03:04:06',
c_money=NULL,
c_numeric10_3=111.444,
c_path='((0,0),(10,10),(15.1,1.1))',
c_pg_lsn='0/8097370',
c_point='(2.2,99.9)',
c_polygon='((1,1),(10,10),(15.1,15.1),(1,1))'::polygon,
c_float4=99.8::float4,
c_smallint=2,
c_smallserial=2,
c_serial=2,
c_text='abc1234',
c_time='00:04:28.237357',
c_timetz='00:04:47.28835+09'::timetz,
c_timestamp='2019-12-14 00:05:21.820347'::timestamp,
c_timestamptz='2019-12-14 00:05:21.820347+09'::timestamptz,
c_tsquery='fat & dog'::tsquery,
c_tsvector='a fat dog'::tsvector,
c_txid_snapshot='674:674:'::txid_snapshot,
c_uuid='a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a12'::uuid,
c_xml='<foo>bar1</foo>'::xml,
c_inta='{1,2,4,5}'::int[]
where id=1;

delete from tb_types;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);


-- Kinds of SQL test

select pg_current_wal_lsn() lsn_1 \gset

truncate tb1;

insert into tb1 values(1,'1'),(2,'2'),(3,'3');

update tb1 set c1='xx' || c1 where id>=2;

INSERT INTO tb1 (id, c1)
VALUES (1, 'new1'), (2, 'new2'), (10, 'new10')
ON CONFLICT (id)
  DO UPDATE SET c1 = EXCLUDED.c1;

delete from tb1 where id <=2;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);


-- Kinds of table test

---- generated common
select pg_current_wal_lsn() lsn_1 \gset

CREATE TABLE tb_generated_common(id int primary key,c1 int GENERATED BY DEFAULT AS IDENTITY,c2 int GENERATED BY DEFAULT AS IDENTITY);

insert into tb_generated_common(id) values(1),(2);
update tb_generated_common set c1=10 where id=1;
delete from tb_generated_common;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

DROP TABLE tb_generated_common;

---- union PK
select pg_current_wal_lsn() lsn_1 \gset

CREATE TABLE tb_unionpk_test(id1 int,id2 text,c1 text,PRIMARY KEY(id1,id2));

insert into tb_unionpk_test values(1,'1','1'),(2,'2','2');
update tb_unionpk_test set c1='xxx' where id1=1 and id2='1';
delete from tb_unionpk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

DROP TABLE tb_unionpk_test;

---- REPLICA IDENTITY(with UK)
select pg_current_wal_lsn() lsn_1 \gset

CREATE TABLE tb_repl_id_uk_test(id1 int not null,id2 text not null,c1 text);
CREATE UNIQUE INDEX tb_repl_id_uk_test_ukidx ON tb_repl_id_uk_test(id1,id2);
 
insert into tb_repl_id_uk_test values(1,'1','1'),(2,'2','2'),(3,'3','3');
update tb_repl_id_uk_test set c1='xxx' where id1=1 and id2='1';
delete from tb_repl_id_uk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

select pg_current_wal_lsn() lsn_1 \gset
ALTER TABLE tb_repl_id_uk_test REPLICA IDENTITY  USING INDEX tb_repl_id_uk_test_ukidx;

insert into tb_repl_id_uk_test values(1,'1','1'),(2,'2','2'),(3,'3','3');
update tb_repl_id_uk_test set c1='xxx' where id1=1 and id2='1';
delete from tb_repl_id_uk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

select pg_current_wal_lsn() lsn_1 \gset
ALTER TABLE tb_repl_id_uk_test REPLICA IDENTITY FULL;

insert into tb_repl_id_uk_test values(1,'1','1'),(2,'2','2'),(3,'3','3');
update tb_repl_id_uk_test set c1='xxx' where id1=1 and id2='1';
delete from tb_repl_id_uk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

DROP TABLE tb_repl_id_uk_test;

---- REPLICA IDENTITY(without UK)
select pg_current_wal_lsn() lsn_1 \gset

CREATE TABLE tb_repl_id_nouk_test(id1 int not null,id2 text not null,c1 text);
 
insert into tb_repl_id_nouk_test values(1,'1','1'),(2,'2','2'),(3,'3','3'),(3,'3','3');
update tb_repl_id_nouk_test set c1='xxx'||c1 where id1=3 and id2='3';
delete from tb_repl_id_nouk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

select pg_current_wal_lsn() lsn_1 \gset
ALTER TABLE tb_repl_id_nouk_test REPLICA IDENTITY FULL;

insert into tb_repl_id_nouk_test values(1,'1','1'),(2,'2','2'),(3,'3','3');
update tb_repl_id_nouk_test set c1='xxx'||c1 where id1=3 and id2='3';
delete from tb_repl_id_nouk_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

DROP TABLE tb_repl_id_nouk_test;

---- TEMP TABLE and UNLOGGED TABLE

select pg_current_wal_lsn() lsn_1 \gset

CREATE TEMP TABLE tb_tmp_test(id1 int,id2 text,c1 text,PRIMARY KEY(id1,id2));

insert into tb_tmp_test values(1,'1','1'),(2,'2','2');
update tb_tmp_test set c1='xxx' where id1=1 and id2='1';
delete from tb_tmp_test;

CREATE UNLOGGED TABLE tb_unlogged_test(id1 int,id2 text,c1 text,PRIMARY KEY(id1,id2));

insert into tb_unlogged_test values(1,'1','1'),(2,'2','2');
update tb_unlogged_test set c1='xxx' where id1=1 and id2='1';
delete from tb_unlogged_test;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);

DROP TABLE tb_unlogged_test;

-- Transaction test

select pg_current_wal_lsn() lsn_1 \gset

BEGIN;
insert into tb1 values(1,'1'),(2,'2');
update tb1 set c1='xxx' where id=1;

select pg_current_wal_lsn() lsn_2 \gset
COMMIT;
select pg_current_wal_lsn() lsn_3 \gset
delete from tb1;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);
select op_type,op_text from wal2sql(NULL,:'lsn_1',:'lsn_2');
select op_type,op_text from wal2sql(NULL,:'lsn_2',:'lsn_3');
select op_type,op_text from wal2sql(NULL,:'lsn_3',NULL);


-- Transaction ROLLBACK test

select pg_current_wal_lsn() lsn_1 \gset

BEGIN;
insert into tb1 values(1,'1'),(2,'2');
update tb1 set c1='xxx' where id=1;
ROLLBACK;


BEGIN;
    INSERT INTO tb1 values(1,'1'),(2,'2');
    SAVEPOINT my_savepoint;
    INSERT INTO tb1 VALUES (3,'3');
    ROLLBACK TO SAVEPOINT my_savepoint;
    INSERT INTO tb1 VALUES (4,'4');
    update tb1 set c1='xxx';

select pg_current_wal_lsn() lsn_2 \gset
COMMIT;
delete from tb1;

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);
select op_type,op_text from wal2sql(NULL,:'lsn_1',:'lsn_2');
select op_type,op_text from wal2sql(NULL,:'lsn_2',NULL);

-- Transaction Concurrent test

select pg_current_wal_lsn() lsn_1 \gset

CREATE EXTENSION IF NOT EXISTS dblink;

SELECT dblink_connect('dbname='|| current_database() ||' port='|| (select setting from pg_settings where name='port') ||' user=' || current_user) ;

BEGIN;
insert into tb1 values(1,'1'),(2,'2');
SELECT dblink_exec('BEGIN');
SELECT dblink_exec($$ insert into tb1 values(3,'3') $$);
update tb1 set c1='xxx';

select pg_current_wal_lsn() lsn_2 \gset
COMMIT;

SELECT dblink_exec($$ update tb1 set c1='yyy' $$);
select pg_current_wal_lsn() lsn_3 \gset
SELECT dblink_exec($$ COMMIT $$);

select op_type,op_text from wal2sql(NULL,:'lsn_1',NULL);
select op_type,op_text from wal2sql(NULL,:'lsn_2',NULL);
select op_type,op_text from wal2sql(NULL,:'lsn_3',NULL);

delete from tb1;

-- timestamp filter test
select now() lsn_1 \gset
insert into tb1 values(1,'tttt'),(2,'tttt');
update tb1 set c1='xxx';
select now() lsn_2 \gset
delete from tb1;

select op_type,op_text from wal2sql_bytime(NULL,:'lsn_1',NULL);
select op_type,op_text from wal2sql_bytime(NULL,:'lsn_1',:'lsn_2');

-- xid filter test
select txid_current() lsn_1 \gset
insert into tb1 values(1,'xxxx'),(2,'xxxx');
update tb1 set c1='xxx';
select txid_current() lsn_2 \gset
delete from tb1;

select op_type,op_text from wal2sql_byxid(NULL,:'lsn_1',NULL);
select op_type,op_text from wal2sql_byxid(NULL,:'lsn_1',:'lsn_2');




